"""
Copyright (c) 2022 Huawei Technologies Co.,Ltd.

openGauss is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:

          http://license.coscl.org.cn/MulanPSL2

THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details.
"""
"""
Case Type   : 故障&可靠性测试
Case Name   : 延迟等待过程中kill -3 各节点进程
Create At   : 2021/03/03
@zou_jialiang0505328126
Description :
    1.设置延迟为5min
    2.切换备节点为级联备
    3.创建表
    4.等待30s，kill -3 主节点
    5.等待30s启动主节点
    6.查询各节点
    7.插入数据;
    8.等待30s kill -3 备节点
    9.等待30s start备节点
    10.查询各节点
    11.更新数据
    12.等待30s，kill -3 级联备
    13.等待30s, start级联备
    14.查询各节点
    15.删除数据
    16.等待30s，kill -3 级联备
    17.等待30s, start级联备
    18.查询各节点
Expect      :
    1.设置成功
    2.切换成功
    3.建表成功
    4.停止成功
    5.重启成功
    6.主节点有数据，备节点5min后同步
    7.插入数据成功
    8.停止备节点成功
    9.重启成功
    10.主节点有数据，备节点5min后同步
    11.更新数据成功
    12.停止成功
    13.启动成功
    14.主节点有数据，备节点5min后同步
    15.删除数据成功
    16.停止成功
    17.启动成功
    18.主节点有数据，备节点5min后同步
History     :
    modified by @wan005 at 2022.06.30:优化为线程同时查询2备机
    modified by n@ningyali at 2023.09.08:中间启停操作耗时未知，增加对这段时间的判断，如大于延迟回放时间则已同步
"""

import os
import re
import time
import unittest
from datetime import datetime

from yat.test import Node
from yat.test import macro

from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.CommonSH import RestartDbCluster
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger


class RecoveryDelay(unittest.TestCase):
    commonshpri = CommonSH('PrimaryDbUser')
    nodes_tuple = ('PrimaryDbUser', 'Standby1DbUser', 'Standby2DbUser')

    @RestartDbCluster(*nodes_tuple)
    def setUp(self):
        self.log = Logger()
        self.log.info(f"-----{os.path.basename(__file__)} start-----")
        self.constant = Constant()
        self.nodes_info = {'pri': {'root_com': Common('PrimaryRoot'),
                                   'root_node': Node(node='PrimaryRoot'),
                                   'user_sh': CommonSH('PrimaryDbUser'),
                                   'node_label': 'pri'},
                           'sta1': {'root_com': Common('Standby1Root'),
                                    'root_node': Node(node='Standby1Root'),
                                    'user_sh': CommonSH('Standby1DbUser'),
                                    'node_label': 'sta1'},
                           'sta2': {'root_com': Common('Standby2Root'),
                                    'root_node': Node(node='Standby2Root'),
                                    'user_sh': CommonSH('Standby2DbUser'),
                                    'node_label': 'sta2'}}
        self.tb_name = 'tb_256'
        self.delay = '5min'

        self.log.info('=====安装ntp服务=======')
        install_cmd = "yum install -y ntp && " \
                      "systemctl enable ntpd && systemctl start ntpd"
        for node in [self.nodes_info['sta1']['root_node'],
                     self.nodes_info['sta2']['root_node']]:
            res = node.sh(install_cmd).result()
            self.log.info(res)

        self.log.info('-----同步集群时间-------')
        for node in [self.nodes_info['sta1']['root_node'],
                     self.nodes_info['sta2']['root_node']]:
            current = self.nodes_info['pri']['root_node'].sh(
                "date \"+%m/%d/%Y %H:%M:%S\"").result()
            self.log.info(current)
            datecmd = f'date -s "{current}";hwclock --systohc;' \
                f'service ntpd stop;ntpdate ntp.api.bz;service ntpd start'
            self.log.info(datecmd)
            result = node.sh(datecmd).result()
            self.log.info(result)

        self.log.info('-----查看主备时间-------')
        cmd = f'source {macro.DB_ENV_PATH};' \
            f'gs_ssh -c "date +\\\"%Y-%m-%d %H:%M:%S\\\"";'
        self.log.info(cmd)
        result = self.nodes_info['pri']['user_sh'].node.sh(cmd).result()
        self.log.info(result)

        self.log.info('-------等待主备一致----------')
        for sh in [self.nodes_info['sta1']['user_sh'],
                   self.nodes_info['sta2']['user_sh']]:
            result = sh.check_data_consistency()
            self.assertTrue(result)

    def check_delay(self, sql, **kill_node):
        self.log.info('-----获取当前时间-----')
        self.log.info('-----执行sql-----')
        start_time = datetime.now()
        self.log.info(start_time)
        result = self.nodes_info['pri']['user_sh'].executDbSql(sql)
        self.log.info(result)
        for expect in ('CREATE', 'INSERT', 'UPDATE', 'DELETE'):
            if expect in str(sql).upper():
                self.assertIn(expect, result)

        self.log.info(f'-----等待30s kill -3 {kill_node["node_label"]}节点-----')
        time.sleep(30)
        result = kill_node['root_com'].kill_pid(kill_node['root_node'], '3')
        self.log.info(result)

        self.log.info(f'-----等待30s启动{kill_node["node_label"]}节点------')
        time.sleep(30)
        tmp = 'primary' if kill_node['node_label'] == 'pri' else 'standby'
        result = kill_node['user_sh'].startDbInstance(tmp)
        self.assertIn(self.constant.RESTART_SUCCESS_MSG, result)

        self.log.info('-----主各节点查询------')
        sql = f"select * from {self.tb_name};"
        pri_res = self.nodes_info['pri']['user_sh'].executDbSql(sql)
        self.log.info(pri_res)

        for n_name, node in self.nodes_info.items():
            if n_name == 'pri':
                continue
            self.log.info('-----获取当前时间-----')
            end_time = datetime.now()
            self.log.info(end_time)
            d_value = (end_time - start_time).seconds
            self.log.info(d_value)

            self.log.info(f'-----备节点{n_name}查询-----')
            sta_res = node['user_sh'].executDbSql(sql)
            self.log.info(sta_res)
            regex_res = re.search('relation.*does not exist', sta_res, re.S)
            if d_value > int(self.delay.strip('min')) * 60:
                self.assertEqual(pri_res, sta_res, '超出延时回放完成')
            else:
                if not regex_res:
                    if kill_node['node_label'] == node['node_label']:
                        self.assertEqual(pri_res, sta_res, '重启回放完成')
                    else:
                        self.assertNotEqual(pri_res, sta_res, '未超出延时未回放')

        d_value = (datetime.now() - start_time).seconds
        self.log.info(d_value)
        if d_value < int(self.delay.strip('min')) * 60:
            self.log.info('等待回放完成，然后再次查询')
            time.sleep(int(self.delay.strip('min')) * 60 - d_value + 30)
            for n_name, node in self.nodes_info.items():
                if n_name == 'pri':
                    continue
                sta_res = node['user_sh'].executDbSql(sql)
                self.log.info(sta_res)
                self.assertEqual(pri_res, sta_res, '超出延时回放完成')
        else:
            self.log.info('回放完成无需等待')

    def test_recoverydelay(self):
        self.log.info(f"修改recovery_min_apply_delay值")
        result = self.nodes_info['pri']['user_sh'].executeGsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG,
            f'recovery_min_apply_delay={self.delay}')
        self.assertTrue(result)

        self.log.info('主机事务提交需要等待主机自身将对应日志刷新到磁盘（本地提交）')
        result = self.nodes_info['pri']['user_sh'].executeGsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG,
            f'synchronous_commit=local')
        self.assertTrue(result)

        sql = f"show synchronous_commit;show recovery_min_apply_delay;"
        for node in self.nodes_info.values():
            sql_res = node['user_sh'].executDbSql(sql)
            self.log.info(f'[{node["node_label"]}] sql_res: {sql_res}')

        self.log.info('------将备节点切换为级联备--------')
        result = self.nodes_info['sta1']['user_sh'].buildStandby(
            '-M cascade_standby')
        self.assertIn(self.constant.REBUILD_SUCCESS_MSG, result)
        result = self.nodes_info['sta1']['user_sh'].exec_refresh_conf()
        self.assertTrue(result)

        self.log.info('-------等待主备一致----------')
        result = self.nodes_info['sta2']['user_sh'].check_data_consistency()
        self.assertTrue(result)
        for i in range(90):
            result = self.nodes_info['pri'][
                'user_sh'].check_cascade_standby_consistency()
            if result:
                break
            time.sleep(20)
        self.assertTrue(result)

        self.log.info('-----创建表-----')
        sql = f"drop table if exists {self.tb_name};" \
            f"create table {self.tb_name}(i int, s char(10));"
        self.check_delay(sql, **self.nodes_info['pri'])

        self.log.info('-----插入数据-----')
        sql = f"insert into {self.tb_name} values (1, 'test');"
        self.check_delay(sql, **self.nodes_info['sta1'])

        self.log.info('-----更新数据-----')
        sql = f"update {self.tb_name} set i=5;"
        self.check_delay(sql, **self.nodes_info['sta2'])

        self.log.info('-----删除数据-----')
        sql = f"delete from {self.tb_name} where i=5;"
        self.check_delay(sql, **self.nodes_info['pri'])

    def tearDown(self):
        self.log.info('--------环境清理--------')
        self.log.info('-----还原recovery_min_apply_delay---------')
        result = self.nodes_info['pri']['user_sh'].executeGsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            'recovery_min_apply_delay=0')
        self.log.info(result)
        self.log.info('-----还原synchronous_commit---------')
        result = self.nodes_info['pri']['user_sh'].executeGsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            'synchronous_commit=on')
        self.log.info(result)

        self.log.info('-------还原集群------')
        self.nodes_info['sta1']['user_sh'].buildStandby('-M standby')
        result = self.nodes_info['sta1']['user_sh'].exec_refresh_conf()
        self.log.info(result)

        self.log.info('----------重启数据库----------')
        result = self.nodes_info['pri']['user_sh'].stopDbCluster()
        self.log.info(result)
        result = self.nodes_info['pri']['user_sh'].startDbCluster()
        self.log.info(result)

        time.sleep(5)
        self.log.info('-------删除表------')
        sql = f"drop table if exists {self.tb_name};"
        result = self.nodes_info['pri']['user_sh'].executDbSql(sql)
        self.log.info(result)
        self.log.info(f"-----{os.path.basename(__file__)} end-----")
